package com.binance.notification.web.utils;

import java.util.Date;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

import com.binance.master.task.AutoTaskConfig;
import org.apache.commons.lang3.time.DateUtils;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;

import com.binance.master.task.TaskUtils;
import com.binance.master.utils.ThreadUtils;
import com.binance.master.utils.WebUtils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
import org.springframework.web.context.support.WebApplicationContextUtils;

import javax.annotation.PostConstruct;

@Slf4j
@Component
public class Worker extends RecursiveAction {

    private static final long serialVersionUID = -8972163997739058624L;
    private static final int CPUS = Runtime.getRuntime().availableProcessors();
    private volatile static ForkJoinPool forkJoinPool = (ForkJoinPool) Executors.newWorkStealingPool(CPUS * 10);
    private int start;
    private int end;
    private int threshold;
    private IJob job;
    private boolean async;
    @Autowired
    private AutoTaskConfig autoTaskConfig;

    public Worker() {

    }

    private Worker(int threshold, int start, int end, IJob job, boolean async) {
        this.threshold = threshold;
        this.start = start;
        this.end = end;
        this.job = job;
        this.async = async;
    }

    /**
     * 提交一个任务，采用工作窃取方式进行操作
     *
     * @param threshold
     * @param start
     * @param end
     * @param job
     * @param async
     */
    public void submitAsWorkStealing(int threshold, int start, int end, IJob job, boolean async) {
        Worker task = new Worker(threshold, start, end, job, async);
        if (task.async) {
            forkJoinPool.submit(task);
        } else {
            try {
                forkJoinPool.submit(task).get();
            } catch (InterruptedException | ExecutionException e) {
                log.warn("", e);
            }
        }
    }

    /**
     * 提交一个一次性任务
     * 
     * @param command 任务
     */
    public void submit(Runnable... command) {
        for (Runnable runnable : command) {
            String taskId = UUID.randomUUID().toString();
            try {
                TaskUtils.addSimpleTask(taskId, new Date(), 0, 0, (trigger, params) -> {
                    runnable.run();
                }, taskId);
            } catch (SchedulerException e) {
                log.warn("", e);
            }
        }
    }

    /**
     * 提交一个在指定延时后执行的一次性任务
     * 
     * @param command
     * @param initialDelay 延迟毫秒执行
     */
    public void schedule(Runnable command, long initialDelay) {
        Date startDate = DateUtils.addMilliseconds(new Date(), (int) (initialDelay < 0 ? 0 : initialDelay));
        String taskId = UUID.randomUUID().toString();
        try {
            TaskUtils.addSimpleTask(taskId, startDate, 0, 0, (trigger, params) -> {
                command.run();
            }, taskId);
        } catch (SchedulerException e) {
            log.warn("", e);
        }
    }

    /**
     * 固定频率执行任务
     * 
     * @param command 任务，当返回false时将停止调度
     * @param initialDelay 延迟毫秒执行
     * @param period 执行间隔，毫秒
     */
    public <V> void scheduleAtFixedRate(Callable<V> command, long initialDelay, long period) {
        Date startDate = DateUtils.addMilliseconds(new Date(), (int) (initialDelay < 0 ? 0 : initialDelay));
        String taskId = UUID.randomUUID().toString();
        try {
            TaskUtils.addSimpleTask(taskId, startDate, period, -1, (trigger, params) -> {
                V value = null;
                try {
                    value = command.call();
                } finally {
                    if (value instanceof Boolean) {
                        if (!((Boolean) value)) {
                            TaskUtils.removeJob((String) params);
                        }
                    }
                }
            }, taskId);
        } catch (SchedulerException e) {
            log.warn("", e);
        }
    }

    @Override
    protected void compute() {
        if (this.end - this.start <= threshold) {
            this.job.execute(start, end);
        } else {
            int middle = (start + end) / 2;
            Worker task1 = new Worker(threshold, start, middle, job, async);
            Worker task2 = new Worker(threshold, middle, end, job, async);
            if (async) {
                task1.fork();
                task2.fork();
            } else {
                invokeAll(task1, task2);
            }
        }
    }


    public interface IJob {

        public void execute(int start, int end);

    }

}
